package javaVersion.learn.checkPoint;

import javaVersion.pojo.SEvent;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;

import javax.annotation.Nullable;

/**
 * @program: myFlink
 * @description: ，具体实现业务流程，另外，代码中窗口的触发时间使 用了event time
 * @author: WincoMa
 * @create: 2020-07-23 15:20
 **/
public class Eg1_FlinkEventTimeAPIChkMain {
    public static void main(String[] args) {
        // TODO: 2020/7/23 环境处理
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        //开启ck，间隔时间为6s
        senv.enableCheckpointing(6000);
        //设置状态后端位置
        StateBackend backend = new FsStateBackend("hdfs://hadoop1:9000/flink-checkpoint/checkpoint/");
        senv.setStateBackend(backend);
        //设置ck
        CheckpointConfig ck = senv.getCheckpointConfig();
        ck.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //设置处理时间为事件的生成时间
        senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // TODO: 2020/7/23 获取数据源，处理逻辑
        DataStreamSource<SEvent> source = senv.addSource(new Eg1_SEventSourceWithChk());

        //对数据源添加水印支持，目的是为了允许延迟
        //指定水印时两种水印类型
        /**
         * Assigner With Periodic Watermarks 周期性水印
         * Assigner With Punctuated Watermarks 间断性水印
         */
        // TODO: 2020/7/23 对数据设置水印
        SingleOutputStreamOperator<SEvent> watermarkDataStream = source.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<SEvent>() {
            @Nullable
            @Override
            public Watermark getCurrentWatermark() {
                //设置水印
                Watermark watermark = new Watermark(System.currentTimeMillis());
                return watermark;
            }

            @Override
            public long extractTimestamp(SEvent element, long previousElementTimestamp) {
                //给每个元素打上时间戳
                long l = System.currentTimeMillis();
                return l;
            }
        });

        // TODO: 2020/7/23 对设置水印后的数据进行处理
        watermarkDataStream.keyBy("id")
                            //对流设置窗口
                            .timeWindow(Time.seconds(4), Time.seconds(1))
                            //指定自定义窗口
                            .apply(new Eg1_WindowStatisticWithChk())
                            .print();
        try {
            senv.execute();
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("执行失败！");
        }
    }
}
